//创建子线程以及子线程的上传下载功能
#include "factory.h"
int upload(int fd_up, char* file_name,off_t off){
	int fd_file = open(file_name, O_RDONLY);
	if(fd_file == -1){
		return -1 ;
	}
    lseek(fd_file,off,SEEK_SET);
	char *read_buf = (char*)malloc(8 * 1024);
	bzero(read_buf, 8 * 1024);
	int nread ;
	while(1){
		nread = read(fd_file, read_buf, 8192);
		if(nread < 8192){
			send(fd_up, &nread, sizeof(int), 0);
			send_n(fd_up, read_buf, nread);
			break ;
		}else{
			send(fd_up, &nread, sizeof(int), 0);
			send_n(fd_up, read_buf, nread);
		}
	}
	int flag = 0 ;
	send(fd_up, &flag, sizeof(flag), 0);
	close(fd_file);
	return 0 ;
}
//带有断点续传功能的下载
int download(int fd_down, char* file_name){
	int fd_file = open(file_name,O_RDWR|O_CREAT,0666);
	ERROR_CHECK(fd_file,-1,"open");
	//lseek(fd_file,0,SEEK_END);
	char* write_buf = (char*)malloc(8192);
	bzero(write_buf, 8192);
	int nwrite ;
	while(1){
		recv_n(fd_down, &nwrite, sizeof(int));
		if(nwrite == 0){
			break ;
		}
		recv_n(fd_down, write_buf, nwrite);
		write(fd_file, write_buf, nwrite);
	}
	close(fd_file);
	return 0 ;
}
int factory_init(Factory_t *pf,int thread_num,int capacity){
    que_init(&pf->que,capacity);//队列初始化
    int ret=pthread_cond_init(&pf->cond,NULL);
    THREAD_ERR_CHECK(ret,"pthread_cond_init");
    pf->pthid=(pthread_t *)calloc(thread_num,sizeof(pthread_t));//为线程id申请空间
    pf->start_flag=0;
    return 0;
}

int factory_start(Factory_t *pf,int thread_num){
    if(!pf->start_flag){//如果线程池没启动
        int i;
        for(i=0;i<thread_num;i++){
            pthread_create(pf->pthid+i,NULL,thread_handle,pf);
        }
        pf->start_flag=1;
    }
}
//释放锁资源
void cleanup(void *p){
    pthread_mutex_unlock((pthread_mutex_t*)p);
}
//子线程流程
void* thread_handle(void *p){
    Factory_t* pf=(Factory_t*)p;
    ptQue_t pq=&pf->que;
    train_state_t t;//等着接md5值
    train_off_t off_train;//等着接偏移值
    pNode_t pcur=NULL;//取队列结点
    while(1){
        pthread_mutex_lock(&pq->mutex);
        pthread_cleanup_push(cleanup,&pq->mutex);
        if(0==pq->que_size){//如果队列为空就睡觉
            pthread_cond_wait(&pf->cond,&pq->mutex);
        }
        que_get(pq,&pcur);//取结点
        pthread_cleanup_pop(1);//直接通过pop来解锁，正常走到此处或者被cancle都会解锁
        if(pcur){//pcur不为NULL代表拿到了结点
            finish_task(pcur,&off_train,&t);
            free(pcur);
            pcur=NULL;
        }
    }
}
void finish_task(pNode_t pN,train_off_t* poff_train,train_state_t* pt){//若拿到上传任务则，拿到下载任务则继续接受
    char query[100]={0};
    char result[100]={0};
    char reference_count[10]={0};
    int ref;
    bzero(poff_train,sizeof(train_off_t));
    bzero(pt,sizeof(train_state_t));
    if(pN->state==GETS){//拿到下载任务则接收一个偏移小火车
        recv_n(pN->new_fd,&poff_train->data_len,4);
        recv_n(pN->new_fd,&poff_train->state,4+sizeof(poff_train->file_off));
        printf("拿到了下载任务,偏移为%ld\n",poff_train->file_off);
        bzero(query,100);
        sprintf(query,"select md5 from FileList where filename = '%s';",pN->filename);
        bzero(result,100);
        search_from_database(query,result);//因为文件存储时实际是以该文件的MD5值作为文件名存储的
        result[strlen(result)-1]=0;//所以要查表拿到该文件的md5值
        upload(pN->new_fd,result,poff_train->file_off);
    }else{//拿到上传任务则接收一个md5码的小火车
        recv_session(pN->new_fd,pt);
        sprintf(query,"select md5 from Md5_File where md5 = '%s';",pt->buf);
        search_from_database(query,result);
        if(strncmp(result,pt->buf,strlen(pt->buf))==0){//比对成功
            bzero(query,100);
            sprintf(query,"select reference_count from Md5_File where md5 = '%s';",pt->buf);
            search_from_database(query,reference_count);
            ref=atoi(reference_count);
            bzero(query,100);
            sprintf(query,"update Md5_File set reference_count = '%d' where md5 = '%s';",ref+1,pt->buf);
            delete_insert_updata(query);
            bzero(query,100);
            sprintf(query,"insert into FileList(precode,filename,filetype,md5sum) values('%d','%s','%s','%s');",pN->code,pN->filename,"f",pt->buf);
            return ;
        }else{//插入一条数据
            bzero(query,100);
            sprintf(query,"insert into Md5_File(md5,reference_count) values('%s','%d');",pt->buf,1);
            delete_insert_updata(query);
            bzero(query,100);
            sprintf(query,"insert into FileList(precode,filename,filetype,md5sum) values('%d','%s','%s','%s');",pN->code,pN->filename,"f",pt->buf);
            
            download(pN->new_fd,pt->buf);//以md5值来作为文件名
        }
    }
}